package xyj.sparrow.scheduler.server.connect;

import lombok.extern.slf4j.Slf4j;
import org.apache.curator.x.discovery.ServiceInstance;
import xyj.sparrow.common.connect.SparrowConnect;
import xyj.sparrow.common.discovery.SparrowServiceInstance;
import xyj.sparrow.discovery.api.SparrowDiscovery;
import xyj.sparrow.discovery.zk.bean.ZkSparrowDiscovery;
import xyj.sparrow.discovery.zk.bean.strategies.ProviderStrategy;
import xyj.sparrow.scheduler.entity.App;
import xyj.sparrow.scheduler.mapper.AppMapper;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.function.Consumer;
import java.util.stream.Collectors;

/**
 * 使用服务发现组件管理连接
 * 主要是选取和分片参数
 *
 * @author 江南小俊
 * @since 2021/6/25 上午7:09
 **/
@Slf4j
public class DiscoveryConnectManager implements ConnectManager {
    @Resource
    private SparrowDiscovery sparrowDiscovery;
    private final ConcurrentHashMap<String, CopyOnWriteArrayList<SparrowConnect>> connectMap =
            new ConcurrentHashMap<>();
    private final ConcurrentHashMap<String, Consumer> consumerMap = new ConcurrentHashMap<>();
    @Resource
    private AppMapper appMapper;

    @Override
    public void addConnect(SparrowConnect connect) {
        if (!connectMap.containsKey(connect.getAppId())) {
            synchronized (connectMap) {
                if (!connectMap.containsKey(connect.getAppId())) {
                    CopyOnWriteArrayList<SparrowConnect> sparrowConnects = new CopyOnWriteArrayList<>();
                    connectMap.put(connect.getAppId(), sparrowConnects);
                }
            }
        }
        connectMap.get(connect.getAppId()).add(connect);
        // 加consumer
        if (!consumerMap.containsKey(connect.getAppId())) {
            synchronized (consumerMap) {
                if (!consumerMap.containsKey(connect.getAppId())) {
                    try {
                        Consumer consumer = new ConnectConsumer(connectMap.get(connect.getAppId()));
                        sparrowDiscovery.addListener("apps/" + connect.getAppId(), ProviderStrategy.ROUND_ROBIN,
                                consumer);
                        consumerMap.put(connect.getAppId(), consumer);
                    } catch (Exception e) {
                        log.error("", e);
                        e.printStackTrace();
                    }
                }
            }
        }
        String[] split = connect.getConnectDesc().split(":");
        try {
            Object register = sparrowDiscovery.register("apps/" + connect.getAppId(), split[0],
                    Integer.parseInt(split[1]), "");
            connect.setPayload(register);
        } catch (Exception e) {
            log.error("", e);
        }


    }

    @Override
    public void removeConnect(SparrowConnect connect) {
        connectMap.forEachValue(connectMap.size(), (value) -> {
            for (SparrowConnect sparrowConnect : value) {
                if (sparrowConnect.getChannel().equals(connect.getChannel())) {
                    value.remove(sparrowConnect);
                    sparrowDiscovery.unregister(Collections.singletonList(sparrowConnect.getPayload().toString()));
                    log.info("remove client success");
                }
            }
        });
    }

    @Override
    public List<String> listConnectDesc(String appId) {
        //这个要从注册中心拉取了
        try {
            List<SparrowServiceInstance> list = sparrowDiscovery.listInstance("apps/" + appId);
            return list.stream().map(item -> item.getAddress() + ":" + item.getPort()).collect(Collectors.toList());
        } catch (Exception e) {
            e.printStackTrace();
        }
        return new ArrayList<>();
    }

    @Override
    public List<SparrowConnect> getConnect(String appId) {
        //获取连接
        CopyOnWriteArrayList<SparrowConnect> sparrowConnects = connectMap.get(appId);
        if (sparrowConnects != null && !sparrowConnects.isEmpty()) {
            //todo 定义策略（第一个\最后一个\广播）  分布策略就挺难
            App app = appMapper.selectById(appId);
            //广播模式就要关注分片参数了connectIndex\connectTotal
            switch (app.getStrategy()) {
                case -1: {
                    log.info("{}:最后一个客户端", app.getAppId());
                    //最后一个
                    List<SparrowConnect> collect =
                            sparrowConnects.stream().filter(sparrowConnect -> sparrowConnect.getConnectIndex() == sparrowConnects.size()).collect(Collectors.toList());
                    return collect == null ? new ArrayList<>() : collect;
                }
                case 0: {
                    log.info("{}:广播全部", app.getAppId());
                    return sparrowConnects;
                }
                case 1: {
                    log.info("{}:第一个客户端", app.getAppId());
                    List<SparrowConnect> collect =
                            sparrowConnects.stream().filter(sparrowConnect -> sparrowConnect.getConnectIndex() == 0).collect(Collectors.toList());
                    return collect == null ? new ArrayList<>() : collect;
                }
            }
            return Collections.singletonList(sparrowConnects.get(0));
        }
        return new ArrayList<>();
    }
}
